import json
from datetime import datetime

from celery import shared_task
from internal.entitry.tickets_entitry import TicketsStstaus, ProxyType
from internal.model import Member, TourGuide, TravelAgency


@shared_task
def rservation(bash_id: str, agency_id: str, lv: str, tickets_bean: str, member: str,grounp_id:str,type:str=ProxyType.CUSTOM.value,secret_id:str=None,secret_key=None):
    from internal.service.tickets_service import TicketsService
    from app.http.module import injector
    from internal.model.tour_guide import TourGuide
    from pkg.sqlalchemy import SQLAlchemy
    from internal.entitry.tickets_entitry import TicketsStstaus
    from internal.model.member import Member
    from internal.model.travel_agency import TravelAgency
    from internal.exception.exception import FailException
    from redis import Redis
    import time

    redis_client = injector.get(Redis)
    tickets_service = injector.get(TicketsService)
    db = injector.get(SQLAlchemy)

    # 解析参数
    member_data = json.loads(member)
    tour_guide = TourGuide(**json.loads(tickets_bean))
    travel_agency = TravelAgency(**json.loads(lv))
    batch_completed_key = f"batch_completed:{bash_id}"
    if redis_client.exists(batch_completed_key):
        print(f"🎯 检测到批次 {bash_id} 已完成，跳过任务执行")
        return f"批次 {bash_id} 已完成，无需处理"

    # 🔥 使用成员级别锁 - 预约并行
    member_lock_key = f"member_lock:{member_data['id']}"
    member_lock = redis_client.lock(member_lock_key, timeout=300, blocking_timeout=5)

    try:
        # 尝试获取成员锁
        if member_lock.acquire():
            print(f"🔒 成功获取成员Redis锁: {member_lock_key}")

            try:
                is_stopped, stop_type = tickets_service.is_stopped(bash_id, agency_id)
                if is_stopped:
                    print(f"🛑 任务开始前检测到{stop_type}级别停止，直接退出")
                    print(f"   批次: {bash_id}, 旅行社: {agency_id}, 成员: {member_data['id']}")

                    # 更新成员状态为停止
                    with db.auto_commit():
                        member_record = db.session.query(Member).filter(
                            Member.id == member_data['id']
                        ).first()
                        if member_record:
                            member_record.status = TicketsStstaus.STOP.value
                            print(f"✅ 已更新成员 {member_data['id']} 状态为 STOP")

                    return f"成员 {member_data['id']} 因{stop_type}级别停止而取消"

                max_retries = 3

                for attempt in range(max_retries):
                    try:
                        print(
                            f"开始处理成员预约: bash_id={bash_id}, 成员={member_data['id']} (尝试 {attempt + 1}/{max_retries})")

                        # 🔥 检查停止状态
                        is_stopped, stop_type = tickets_service.is_stopped(bash_id, agency_id)
                        if is_stopped:
                            print(f"🛑 锁定前检测到{stop_type}级别停止，退出处理")
                            with db.auto_commit():
                                member_record = db.session.query(Member).filter(
                                    Member.id == member_data['id']
                                ).first()
                                if member_record:
                                    member_record.status = TicketsStstaus.STOP.value
                            return f"成员 {member_data['id']} 因{stop_type}级别停止而取消"



                        # 执行预约
                        print(f"🎯 执行预约: {member_data['id']}")

                        # 重新获取成员用于预约
                        with db.auto_commit():
                            member_for_reservation = db.session.query(Member).filter(
                                Member.id == member_data['id']
                            ).first()

                            if not member_for_reservation:
                                print(f"❌ 成员 {member_data['id']} 不存在")
                                return f"成员 {member_data['id']} 不存在"
                            if member_for_reservation.status in [TicketsStstaus.CANCELED.value,
                                                                 TicketsStstaus.STOP.value]:
                                print(
                                    f"⏹️ 成员 {member_for_reservation.id} 当前状态为 {member_for_reservation.status}，跳过预约")
                                return f"成员 {member_for_reservation.id} 状态为 {member_for_reservation.status}，不执行预约"

                            if member_for_reservation.status not in [TicketsStstaus.PADDING.value,
                                                                     TicketsStstaus.ERROR.value]:
                                print(f"⚠️ 成员 {member_for_reservation.id} 状态 {member_for_reservation.status} 不支持预约，跳过")
                                return f"成员 {member_for_reservation.id} 状态不支持预约"
                            current_time = datetime.now()
                            deadline_time = current_time.replace(hour=21, minute=59, second=0, microsecond=0)
                            if current_time >= deadline_time:
                                print(f"⏰ 执行预约前检测到已超过截止时间 21:59")
                                if member_for_reservation.status == TicketsStstaus.ERROR.value:
                                    member_for_reservation.status = TicketsStstaus.LAST_ERROR.value
                                    print(f"✅ 已更新成员 {member_for_reservation.id} 状态为 LAST_ERROR")
                                return f"成员 {member_for_reservation.id} 已超时"

                            try:
                                result = tickets_service.rservation_tickets(
                                    bash_id, tour_guide, travel_agency, member_for_reservation,type,secret_id,secret_key
                                )
                            except FailException  as e:
                                if "已超过截止时间" in str(e) or "达到最大重试次数" in str(e):
                                    print(f"⏰ rservation_tickets 内部检测到超时")
                                    if member_for_reservation.status == TicketsStstaus.ERROR.value:
                                        member_for_reservation.status = TicketsStstaus.LAST_ERROR.value
                                        print(f"✅ 已更新成员 {member_for_reservation.id} 状态为 LAST_ERROR")
                                    return f"成员 {member_for_reservation.id} 预约超时"

                            if result:
                                print(f"🎉 成员 {member_for_reservation.id} 预约成功")
                                member_for_reservation.status = TicketsStstaus.RESERVATION_SUCCESS.value
                                print(f"✅ 成功更新成员 {member_for_reservation.id} 状态为 RESERVATION_SUCCESS")
                                # 🔥 外层事务提交：预约成功状态已保存

                        # 🔥 在外层事务提交后执行删除逻辑
                        if result:
                            # 检查重复旅行社
                            same_name_count = db.session.query(TravelAgency).filter(
                                TravelAgency.bash_id == bash_id,
                                TravelAgency.grounp_id == grounp_id,
                                TravelAgency.travel_agency_name == travel_agency.travel_agency_name
                            ).count()

                            is_duplicate_agency = same_name_count > 1

                            if is_duplicate_agency:
                                print(f"⚠️ 当前成员成功，检查是否需要删除其他重复成员")

                                # 查找同一个旅行社的其他预约中成员
                                members_to_delete = (db.session.query(Member)
                                .filter(
                                    Member.bash_id == bash_id,
                                    Member.status == TicketsStstaus.ERROR.value,
                                    Member.grounp_id == grounp_id,
                                    Member.id != member_for_reservation.id
                                )).all()

                                if members_to_delete:
                                    print(f"🔍 找到 {len(members_to_delete)} 个其他预约中成员需要删除")

                                    # 🔥 在独立事务中执行删除
                                    with db.auto_commit():
                                        for member_to_delete in members_to_delete:
                                            print(f"🗑️ 删除重复成员 {member_to_delete.id}")
                                            acy = member_to_delete.get_agency
                                            db.session.delete(acy)
                                            db.session.delete(member_to_delete)
                                            db.session.flush()
                                    print(f"✅ 成功删除 {len(members_to_delete)} 个重复成员")
                                else:
                                    print(f"✅ 没有其他预约中成员需要删除")

                            # 检查批次完成状态
                            _check_batch_completion(bash_id, db)
                            return f"成员 {member_for_reservation.id} 预约成功"
                        else:
                            print(f"❌ 成员 {member_for_reservation.id} 预约返回 False")
                            return f"成员 {member_for_reservation.id} 预约失败"
                    except Exception as e:
                        # 死锁重试逻辑...
                        if attempt < max_retries - 1:
                            time.sleep(0.1 * (attempt + 1))
                            continue
                        else:
                            raise FailException(f"成员 {member_data['id']} 处理失败，达到最大重试次数")

            finally:
                # 释放成员锁
                member_lock.release()
                print(f"🔓 释放成员Redis锁: {member_lock_key}")
        else:
            print(f"⏩ 成员 {member_data['id']} 正在被其他任务处理，跳过")
            return f"成员 {member_data['id']} 正在处理中"

    except Exception as e:
        # 确保异常时也释放锁
        if member_lock.locked():
            member_lock.release()
        raise e

    return f"成员 {member_data['id']} 处理完成"


def _check_batch_completion(bash_id: str, db):
    """检查批次是否完成（所有成员都有最终状态），如果完成则设置停止标志"""
    from app.http.module import injector
    from redis import Redis

    redis_client = injector.get(Redis)
    db.session.expire_all()
    with db.auto_commit():
        all_members = db.session.query(Member).filter(
            Member.bash_id == bash_id
        ).all()

        if not all_members:
            print(f"⚠️ 批次 {bash_id} 没有有效成员")
            return True

        # 统计各种状态
        success_count = 0
        final_state_count = 0  # 最终状态计数

        print(f"🔍 批次 {bash_id} 成员状态详情:")
        for i, m in enumerate(all_members):
            if m.status == TicketsStstaus.RESERVATION_SUCCESS.value:
                status_text = "✅成功"
                success_count += 1
                final_state_count += 1
            elif m.status in [TicketsStstaus.LAST_ERROR.value, TicketsStstaus.STOP.value,
                              TicketsStstaus.CANCELED.value]:
                status_text = "⏹️终态"  # 最终失败或停止
                final_state_count += 1
            elif m.status == TicketsStstaus.ERROR.value:
                status_text = "🔄错误"  # 可能还会重试
                # 不计入final_state_count
            else:  # PADDING等其他状态
                status_text = "⏳处理中"
                # 不计入final_state_count

            print(f"  成员{i + 1}: ID={m.id}, Status={status_text}({m.status})")

        total_count = len(all_members)
        print(f"📊 批次 {bash_id} 进度: 成功={success_count}, 终态={final_state_count}, 总数={total_count}")

        # 🔥 新逻辑：所有成员都达到最终状态（成功或最终失败）
        all_members_final = (final_state_count == total_count)

        if all_members_final:
            # 🔥 关键：设置批次完成标志，让其他任务知道应该停止
            batch_completed_key = f"batch_completed:{bash_id}"
            redis_client.setex(batch_completed_key, 3600, "true")  # 1小时过期
            print(f"🚩 已设置批次完成标志: {batch_completed_key}")

            if success_count == total_count:
                # 所有成员都成功
                print(f"🎊 批次 {bash_id} 所有成员都预约成功！")
                return True
            else:
                # 有成员最终失败，批次完成但未完全成功
                print(f"🏁 批次 {bash_id} 完成：{success_count}/{total_count} 成功，但有最终失败成员")
                return True
        else:
            print(f"⏳ 批次 {bash_id} 还有 {total_count - final_state_count} 个成员在处理中")
            return False